package com.myCompany.edu.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.myCompany.edu.realtime.app.BaseAppV1;
import com.myCompany.edu.realtime.common.Constant;
import com.myCompany.edu.realtime.util.AtguiguUtil;
import com.myCompany.edu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.HashMap;
import java.util.Map;


/**
 * @ClassName：Dwd_BaseLogApp
 * @Author: xxx
 * @Date: 2022/10/20 15:24
 * @Description:
 */
public class Dwd_BaseLogApp extends BaseAppV1 {

    private final String PAGE = "page";
    private final String DISPLAY = "display";
    private final String ERR = "err";
    private final String ACTION = "action";
    private final String START = "start";
    private final String APPVIDEO = "appvideo";

    public static void main(String[] args) {
        new Dwd_BaseLogApp().init(
                2001,
                2,
                "Dwd_BaseLogApp",
                Constant.TOPIC_ODS_BASE_LOG
        );
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        SingleOutputStreamOperator<JSONObject> etledStream = etl(stream);
        SingleOutputStreamOperator<JSONObject> validatedStream = validateNewOrOld(etledStream);
        Map<String, DataStream<JSONObject>> streams = splitStream(validatedStream);
        writeToKafka(streams);
    }

    private void writeToKafka(Map<String, DataStream<JSONObject>> streams) {
        streams.get(START)
                .map(JSONAware::toJSONString)
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_START));

        streams.get(DISPLAY)
                .map(JSONAware::toJSONString)
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_DISPLAY));

        streams.get(ACTION)
                .map(JSONAware::toJSONString)
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ACTION));

        streams.get(ERR)
                .map(JSONAware::toJSONString)
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));

        streams.get(PAGE)
                .map(JSONAware::toJSONString)
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PAGE));

        streams.get(APPVIDEO)
                .map(JSONAware::toJSONString)
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_APPVIDEO));
    }

    private Map<String, DataStream<JSONObject>> splitStream(SingleOutputStreamOperator<JSONObject> validatedStream) {
        OutputTag<JSONObject> displayTag = new OutputTag<JSONObject>("display") {};
        OutputTag<JSONObject> actionTag = new OutputTag<JSONObject>("action") {};
        OutputTag<JSONObject> errTag = new OutputTag<JSONObject>("err") {};
        OutputTag<JSONObject> pageTag = new OutputTag<JSONObject>("page") {};
        OutputTag<JSONObject> AppVideoTag = new OutputTag<JSONObject>("appvideo") {};


        SingleOutputStreamOperator<JSONObject> startStream = validatedStream
                .process(new ProcessFunction<JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject obj,
                                               Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject common = obj.getJSONObject("common");
                        JSONObject page = obj.getJSONObject("page");
                        Long ts = obj.getLong("ts");

                        JSONObject start = obj.getJSONObject("start");
                        if (start != null) {
                            out.collect(obj);
                        } else {

                            JSONArray displays = obj.getJSONArray("displays");
                            if (displays != null) {
                                //一条数据中, 有可能多个曝光, 最好一个曝光一条数据
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject display = displays.getJSONObject(i);
                                    display.putAll(common); // common中的的kv直接添加到了display中
                                    display.putAll(page);
                                    display.put("ts", ts);
                                    ctx.output(displayTag, display);
                                }
                                obj.remove("displays");
                            }


                            JSONArray actions = obj.getJSONArray("actions");
                            if (actions != null) {
                                for (int i = 0; i < actions.size(); i++) {
                                    JSONObject action = actions.getJSONObject(i);
                                    action.putAll(common); // common中的的kv直接添加到了action中
                                    action.putAll(page);
                                    // action有自己的ts, 所以不需要外面的ts
                                    //action.put("ts", ts);
                                    ctx.output(actionTag, action);
                                }
                                obj.remove("actions");
                            }

                            JSONObject err = obj.getJSONObject("err");
                            if (err != null) {
                                ctx.output(errTag, obj);
                                obj.remove("err");
                            }

                            JSONObject appVideo = obj.getJSONObject("appVideo");
                            if (appVideo != null) {
                                appVideo.putAll(common);
                                if (err != null) {
                                    appVideo.putAll(err);
                                }
                                appVideo.put("ts", ts);
                                ctx.output(AppVideoTag, appVideo);
                                obj.remove("appVideo");
                            }
                            // 4. page
                            ctx.output(pageTag, obj);

                        }

                    }
                });

        DataStream<JSONObject> displayStream = startStream.getSideOutput(displayTag);
        DataStream<JSONObject> actionStream = startStream.getSideOutput(actionTag);
        DataStream<JSONObject> errStream = startStream.getSideOutput(errTag);
        DataStream<JSONObject> pageStream = startStream.getSideOutput(pageTag);
        DataStream<JSONObject> appVideoStream = startStream.getSideOutput(AppVideoTag);

        Map<String, DataStream<JSONObject>> streams = new HashMap<String, DataStream<JSONObject>>();
        streams.put(PAGE, pageStream);
        streams.put(ACTION, actionStream);
        streams.put(DISPLAY, displayStream);
        streams.put(ERR, errStream);
        streams.put(START, startStream);
        streams.put(APPVIDEO, appVideoStream);
        return streams;
    }


    private SingleOutputStreamOperator<JSONObject> validateNewOrOld(
            SingleOutputStreamOperator<JSONObject> etledStream) {
        return etledStream
                .keyBy(obj -> obj.getJSONObject("common").getString("mid"))
                .map(new RichMapFunction<JSONObject, JSONObject>() {

                    private ValueState<String> firstVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        firstVisitDateState = getRuntimeContext()
                                .getState(new ValueStateDescriptor<String>("firstVisitDate", String.class));
                    }

                    @Override
                    public JSONObject map(JSONObject obj) throws Exception {
                        JSONObject common = obj.getJSONObject("common");
                        String isNew = common.getString("is_new");
                        Long ts = obj.getLong("ts");
                        String today = AtguiguUtil.toDate(ts);
                        // 新老客户纠正
                        // 1. 首次访问日期
                        String firstVisitDate = firstVisitDateState.value();
                        if (firstVisitDate == null) {
                            if ("1".equals(isNew)) {
                                // 这个用户没有状态, 并且标记还是行用户, 这这个用户应该是第一天的第一次访问
                                // 状态中就存储今天. 更新hi下状态
                                firstVisitDateState.update(today);
                            } else {
                                // 用户是老用户, 但是没有状态. 为了防止以后有问题, 把状态更新成昨天
                                String yesterday = AtguiguUtil.toDate(ts - 24 * 60 * 60 * 1000);
                                firstVisitDateState.update(yesterday);
                            }
                        } else {
                            if ("1".equals(isNew) && !today.equals(firstVisitDate)) {
                                // 状态不为空, 并且今天还和状态不一样, 需要对is_new纠正
                                common.put("is_new", "0");
                            }
                        }

                        return obj;
                    }
                });
    }


    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        try {
                            JSON.parseObject(s);
                        } catch (Exception e) {
                            System.out.println("你的数据格式不是一个有效的json格式: " + s);
                            // 如果有异常,表示json格式不对, 方法false, 数据不要了
                            return false;
                        }
                        // 去掉空白字符之后, 判断长度应该大于0
                        return s.trim().length() > 0;
                    }
                })
                .map(JSON::parseObject);
    }
}
